import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectInputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;




class SqlCon{
	
	private Connection conn = null;
	
	public  void connect(){
		try {
			String connString = "jdbc:mysql://datamining-node01.cs.fiu.edu:33061/pdp?user=pdp&password=pdp";
			Class.forName ("com.mysql.jdbc.Driver").newInstance ();
			setConn(DriverManager.getConnection (connString));
			
		} catch (Exception e) {
			// TODO: handle exception
			System.err.println("Falid:"+e);
		}
	}
	public Connection getConn() {
		return conn;
	}
	public void setConn(Connection conn) {
		this.conn = conn;
	}
}


class Convert{
	private Connection con; 
	private String tblName;
	private String hTblName;
	private byte [] TBLNAME;
	private static byte [] COLFAM_ATT = Bytes.toBytes("att");
	private static byte [] COLFAM_CHAR = Bytes.toBytes("char_id");

	
	private int colnums;
	private String rid;
	private String att_key;
	private String att_value;
	private String col_key;
	
	private Configuration conf;
	private HBaseAdmin admin;
	private HTableDescriptor desc;
	private HTable htable;
	
	Convert(Connection con,String tname,String hname)
	{
		this.tblName = tname;
		this.hTblName = hname;
		this.TBLNAME = Bytes.toBytes(hTblName);
		this.con = con;
	}
	
	private void hbaseInit() throws Exception{
		conf = HBaseConfiguration.create();
		admin = new HBaseAdmin(conf);
		
		if(admin.tableExists(TBLNAME)){
			if(admin.isTableDisabled(TBLNAME)){
				admin.enableTable(TBLNAME);
			}
		
			desc = admin.getTableDescriptor(TBLNAME);
			
		}
		else
		{
			desc = new HTableDescriptor(TBLNAME);
			desc.addFamily(new HColumnDescriptor(COLFAM_ATT));
			desc.addFamily(new HColumnDescriptor(COLFAM_CHAR));
			admin.createTable(desc);
		}
		
		
		htable = new HTable(conf, desc.getName());
		
		//Put put = new Put(ROW1);
		//put.add(COLFAM1, QUA1, Bytes.toBytes("val1"));
		//htable.put(put);		
	}
	
	
	
	public void toHbase(Map<String,String> typeS) throws Exception{
		//con.setAutoCommit(false);
		Statement stmt =  con.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY,
	              java.sql.ResultSet.CONCUR_READ_ONLY);
		
	    stmt.setFetchSize(Integer.MIN_VALUE);
		
		
	//	KvObject kvobject= null;
		Put put = null;
	
		
		System.out.println("Begin");
		
		
		String qString = "Select * from "+tblName;
		
		ResultSet res =  stmt.executeQuery(qString);//LIMIT 0,10
	

		ResultSetMetaData metadata =  res.getMetaData();
		colnums = metadata.getColumnCount();
	
		hbaseInit();
		
		
		int cnt =0;
		while(res.next()){
			System.out.println(cnt++);

			for(int i= 1;i<=colnums;i++){
				String colType = typeS.get(metadata.getColumnName(i));
			
				if(colType.equals(CONSTANT.KEY)){
					rid = res.getString(i);
					//put = new Put(Bytes.toBytes(rid));
					continue;
				}
				else if(colType.equals(CONSTANT.ATT)){
					att_key = metadata.getColumnName(i);
					att_value = res.getString(i);
					put = new Put(Bytes.toBytes(rid+":"+att_key));
					put.add(COLFAM_ATT, Bytes.toBytes(att_key), Bytes.toBytes(att_value));
					htable.put(put);
				}
				else if(colType.equals(CONSTANT.ATT_KEY)){
					//col_key = res.getString(i);
					continue;
				}
				else{
					att_key =metadata.getColumnName(i);
					col_key =  res.getString(colType);
					att_value = res.getString(i);
					put = new Put(Bytes.toBytes(rid+":"+col_key));
					put.add(COLFAM_CHAR, Bytes.toBytes(att_key), Bytes.toBytes(att_value));
					htable.put(put);
					//kvobject.add(metadata.getColumnName(i),att_value);
				}
				
				//System.out.println("att_key:"+ att_key + "att_value:"+att_value);
				
			}	
			
		}
		 res.close();
         htable.close();
		
		
	}
	
}

public class DataConverter {

	/**
	 * @param args
	 * @throws Exception 
	 */	
	
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		
		if(args.length!=2)
		{
			System.err.println("Wrong Arguments.");
		}
		String tblName = args[0];
		String htblName = args[1];
		
		//System.out.println(tblName+htblName);
		SqlCon sqlcon = new SqlCon();
		sqlcon.connect();
		Connection con = sqlcon.getConn(); 
		
		Convert convert = new Convert(con,tblName,htblName);
		//ArrayList<String> typeS = new ArrayList<>();
		Map <String,String> typeS = new HashMap<String, String>();
		
		
	   typeS.put("SUBLOT_ID", CONSTANT.KEY);
	   typeS.put("LOT_ID", CONSTANT.ATT);
	   typeS.put("SLOT_NO", CONSTANT.ATT);
	   typeS.put("GRADE", CONSTANT.ATT);
	   typeS.put("LOSS_CODE", CONSTANT.ATT);
	   typeS.put("LOT_ID_1", CONSTANT.ATT);
	   typeS.put("FLOW", CONSTANT.ATT);
	   typeS.put("OPER", CONSTANT.ATT);
	   typeS.put("EQUIP_ID", CONSTANT.ATT);
	   typeS.put("TRANS_TIME", CONSTANT.ATT);
	   typeS.put("CHAR_ID", CONSTANT.ATT_KEY);
	   typeS.put("VALUE_TBL", "CHAR_ID");
	   typeS.put("RESV_FIELD3", "CHAR_ID");
	   typeS.put("RESV_FIELD4", "CHAR_ID");
	   typeS.put("RESV_FIELD5", "CHAR_ID");
	   
	   convert.toHbase(typeS);
	   //convert.getInfo();
		
				
		
	}

}
